package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.api.R;
import com.heima.common.cache.CacheService;
import com.heima.common.constants.ScheduleConstants;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

@Service
@Transactional
@Slf4j
public class TaskServiceImpl implements TaskService {
    @Autowired
    private TaskinfoMapper taskinfoMapper;

    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;

    @Autowired
    private CacheService cacheService;

    /**
     * todo 目前的场景：用户再提交自媒体文章到数据库后  --》任务的添加
     * 通用接口
     * 添加延迟任务到数据库和redis 中
     * @param task  需要添加的延迟任务  taskDto
     * @return  延迟任务的ID
     */
    @Override
    public long addTask(Task task) {
        //todo 1.把任务数据存储到任务库
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(task,taskinfo);
        taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
        taskinfoMapper.insert(taskinfo);
        //todo 1.1 存储任务日志表 (记录任务日志，仅此而已)
        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(taskinfo,taskinfoLogs);
        taskinfoLogs.setVersion(1);//版本  乐观锁  保证并发安全
        taskinfoLogs.setStatus(ScheduleConstants.SCHEDULE);
        taskinfoLogsMapper.insert(taskinfoLogs);
        //todo 回显任务ID
        task.setTaskId(taskinfo.getTaskId());

        //调用存储redis方法即可
        addTaskToCache(task);

        return taskinfo.getTaskId();
    }


    private void addTaskToCache(Task task){
        //todo 1.5 记得区分任务的类型，和优先级 （redis 的key 用不同key来存储）  1_1
        String key = task.getTaskType() +"_" + task.getPriority();
        //todo 2.拆分任务，判断文章的发布时间，进行任务缓存的存储
        long futureTime = System.currentTimeMillis() + (5*60*1000);
        //todo 2.1  判断执行时间是否 <= 当前时间  立即需要执行延迟任务
        if(task.getExecuteTime() <= System.currentTimeMillis()){  //11:25
            //存储到List队列吧
            cacheService.lLeftPush(ScheduleConstants.TOPIC + key, JSON.toJSONString(task));
        }else if(task.getExecuteTime() <= futureTime){
            //todo 2.2  判断执行时间是否 <= 当前时间+5分钟  未来需要执行任务
            //存储到未来队列
            cacheService.zAdd(ScheduleConstants.FUTURE + key,JSON.toJSONString(task),task.getExecuteTime());
        }
    }


    /**
     * 取消任务
     * @param taskId
     * @return
     */
    @Override
    public boolean cancelTask(long taskId) {
        //0.参数校验
        if(taskId == 0){
            return false;
        }
        //1.取消数据库的任务
        //todo 1.1 修改任务日志中的任务状态
        TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);
        taskinfoLogs.setStatus(ScheduleConstants.CANCELLED);
        taskinfoLogsMapper.updateById(taskinfoLogs);
        //todo 1.2 删除任务表中的任务数据
        taskinfoMapper.deleteById(taskId);
        //2.取消redis队列中的任务
        //封装一个Task对象
        Task task = new Task();
        BeanUtils.copyProperties(taskinfoLogs,task);
        task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());
        //todo 2.1 得到key
        String key = taskinfoLogs.getTaskType() +"_" + taskinfoLogs.getPriority();
        if(taskinfoLogs.getExecuteTime().getTime() <= System.currentTimeMillis()){
            //代表需要进行立即删除  （立即队列）
            cacheService.lRemove(ScheduleConstants.TOPIC+key,0,JSON.toJSONString(task));
        }else{
            //代表需要删除未来队列的数据
            cacheService.zRemove(ScheduleConstants.FUTURE + key ,JSON.toJSONString(task));
        }
        return true;
    }

    /**
     * todo 从立即消费队列中取出任务
     */
    @Override
    public Task poll(int type, int priority) {
        Task task = null;
        //todo 拼接redis的key 值
        String key = type + "_" + priority;
        //todo redis 出栈
        String taskJson = cacheService.lRightPop(ScheduleConstants.TOPIC + key);
        //todo 判断是否取到任务
        if(StringUtils.isNotBlank(taskJson)){
            task = JSON.parseObject(taskJson, Task.class);
            //todo 修改任务日志状态为执行中（1），删除任务表的数据
            //todo 修改任务日志
            TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(task.getTaskId());
            taskinfoLogs.setStatus(ScheduleConstants.EXECUTED);
            taskinfoLogsMapper.updateById(taskinfoLogs);
            //todo 删除任务表的数据
            taskinfoMapper.deleteById(task.getTaskId());
        }
        return task;
    }


    /**
     * todo 把未来队列满足的数据存储到立即消费队列
     * todo 思考： 1.每秒执行一次 (定时任务)   springtask(spring内置的任务调度框架)     Qz(阿帕奇家)    xxl-job(分布式任务调度框架)
     * @Scheduled(cron = "") 代表此方法执行定时执行   cron表达式 方法执行时间（频率）
     */
    @Scheduled(cron = "*/1 * * * * ?")
    public void refresh(){
        //todo 0.查询出未来队列里面所有的任务相关的key值
        Set<String> keys = cacheService.scan(ScheduleConstants.FUTURE+"*");
        //todo 1. 遍历所有的key
        for (String key : keys) {
            //todo 1.1查询未来队列中的满足的数据 (1_1)代表文章审核业务的任务，但是有很多任务 1-2  1-3  1-4  250-250
            Set<String> strings = cacheService.zRangeByScore(key, 0, System.currentTimeMillis());

            //todo 1.2. 根据未来队列key获得消费队列的key
            String topicKey = ScheduleConstants.TOPIC + key.split(ScheduleConstants.FUTURE)[1];
            //todo 2.把未来队列满足的数据存储到消费队列中
            //cacheService.lLeftPushAll(topicKey ,strings);
            //cacheService.zRemoveRangeByScore(key, 0, System.currentTimeMillis());
            //todo 3.使用redis的管道技术来优化代码  (次方法  1.批量添加 2.批量删除)
            if(strings.size() > 0){
                cacheService.refreshWithPipeline(key,topicKey,strings);
            }
        }
        System.out.println("我执行了.............");
    }


    @Autowired
    private RedisTemplate redisTemplate;

    @Scheduled(cron = "0/30 * * * * ? ")
    public void sendYHJ(){
        Boolean flag = redisTemplate.opsForValue().setIfAbsent("yhjlock", "aa");
        if(flag){
            //开始锁续期  -- 子线程来实现
            Thread thread = new Thread(()->{
                while(true){
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    redisTemplate.expire("yhjlock",2, TimeUnit.SECONDS);
                }
            });
            thread.setDaemon(true); //开启守护线程， 备胎线程
            thread.start(); //开启线程
            System.out.println("发送5张优惠卷...."+new Date());
            //释放锁
            redisTemplate.delete("yhjlock");
        }
    }


    /**
     * todo 每隔5分钟从数据库查询任务，存储到redis队列中
     * 问题： 完成了吗？
     */
    @Scheduled(cron = "0 */5 * * * ?")
    //@Scheduled(cron = "0/30 * * * * ?")
    public void relaodDb(){
        //todo 0.清除 redis 中的队列数据
        Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");
        Set<String> topicKeys = cacheService.scan(ScheduleConstants.TOPIC + "*");
        cacheService.delete(futureKeys);
        cacheService.delete(topicKeys);
        //todo 1.查询时间再5分钟内的任务数据
        Date queryDate = new Date(new Date().getTime() + (5 * 60 * 1000));
        LambdaQueryWrapper<Taskinfo> wrapper = new LambdaQueryWrapper<>();
        wrapper.le(Taskinfo::getExecuteTime,queryDate);
        //todo tasks 代表的是<=5分钟内的所有没有消费的数据
        List<Taskinfo> tasks = taskinfoMapper.selectList(wrapper);
        //todo 1.1 判断tasks是否存在
        if(tasks != null && tasks.size() > 0){
            //todo 2.根据查询到的任务的时间开始分配任务去哪个Redis队列
            for (Taskinfo taskinfo : tasks) {
                Task task = new Task();
                BeanUtils.copyProperties(taskinfo,task);
                task.setExecuteTime(taskinfo.getExecuteTime().getTime());
                addTaskToCache(task);
            }
        }
        System.out.println("db定时任务执行成功.....");
    }


    @Scheduled(cron = "0/30 * * * * ?")
    public void sendJ(){
        System.out.println("发放5张优惠卷........");
    }
}
